package com.chief.state;


import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Iterator;
import java.util.Map;

public class KeyedStateTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> streamSource = env.socketTextStream("hadoop55", 7777);

        //一直使用Tuple3,这次我们使用POJO

        SingleOutputStreamOperator<Data> flatMap = streamSource.keyBy(s -> s.split(",")[0]).flatMap(new MyKeyedFlatFunction(33.33));

        flatMap.print();

        env.execute("KeyedStateTest");
    }

    private static class MyKeyedFlatFunction extends RichFlatMapFunction<String, Data> {

        //算子状态
        private final Double level;

        //存储大于level的个数
        private ValueState<Integer> total;

        //存储每个人对应的总和
        private MapState<String, Double> mapState;

        private ReducingState<Data> reducingState;

        private ValueState<Data> betweenState;


        @Override
        public void close() throws Exception {
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            total = getRuntimeContext().getState(new ValueStateDescriptor<>("total", Integer.class));
            mapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("mapState", String.class, Double.class));
            reducingState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<>("reduceState", new ReduceFunction<Data>() {
                @Override
                public Data reduce(Data value1, Data value2) throws Exception {
                    if (value1.getScore().compareTo(value2.getScore()) > 0) {
                        return value1;
                    }
                    return value2;
                }
            }, Data.class));

            betweenState = getRuntimeContext().getState(new ValueStateDescriptor<>("betweenState", Data.class));
        }

        public MyKeyedFlatFunction(Double v) {
            level = v;
        }

        @Override
        public void flatMap(String value, Collector<Data> out) throws Exception {
            String[] split = value.split(",");
            Data data = new Data(split[0], DateUtils.parseDate(split[1], "yyyy-MM-dd HH:mm:ss"), Double.valueOf(split[2]));

            reducingState.add(data);
            System.out.println("reducingState=" + reducingState.get());

            Data lastData = betweenState.value();
            if (null != lastData) {
                if (lastData.getScore().compareTo(data.getScore()) > 0) {
                    System.out.println("lastData greater than this");
                    out.collect(data);
                }
            }
            betweenState.update(data);

            Integer cnt = this.total.value();
            if (level.compareTo(Double.valueOf(split[2])) > 0) {
                if (null == cnt) {
                    cnt = 0;
                }
                cnt++;
                total.update(cnt);

                System.out.println("total=" + total.value());

                Double mapTotal = mapState.get(split[0]);
                if (null == mapTotal) {
                    mapTotal = 0d;
                }
                mapState.put(split[0], mapTotal + Double.valueOf(split[2]));

                Iterator<Map.Entry<String, Double>> iterator = mapState.iterator();
                while (iterator.hasNext()) {
                    Map.Entry<String, Double> next = iterator.next();
                    System.out.println("mapState=" + next.getKey() + "-->" + next.getValue());
                }

                out.collect(data);
            }
        }
    }
}
